[译]Spark调优教程中文版

发表于 | 分类于 |

最近用到Spark比较多，于是打算深入学习下Spark的调优。我谷歌下发现网上调优的经验还是比较散乱，所以决定将官方文档翻译下，为国内开源社区贡献一点力量。另外不得不吐槽翻译真是个体力活，90%的时间都花费在IO上了。

正文

由于Spark基于内存计算的特性，集群的任何资源都可以成为Spark程序的瓶颈:CPU，网络带宽，或者内存。通常，如果内存容得下数据，瓶颈会是网络带宽。不过有时你同样需要做些优化，例如将RDD以序列化到磁盘，来降低内存占用。这个教程会涵盖两个主要话题：数据序列化，它对网络性能尤其重要并可以减少内存使用，以及内存调优。另外我们也会简述几个小一点的话题。

数据序列化

序列化在分布式应用中扮演着重要的角色。序列化对象耗时长，或者占据大量空间的格式，会显著地拖慢计算。通常这会是进行Spark应用调优时你需要做的第一件事。Spark的目标是找到易用性(允许你在操作中使用任何Java类型)以及性能间的平衡。它提供了两个序列化库:

  • Java序列化: 默认情况下，Spark使用Java的 ObjectOutputStream 框架序列化对象，可以处理你创建的实现了 java.io.Serializable 接口的所有Class。你也可以通过继承 java.io.Externalizable 来更紧密地控制序列化性能。Java序列化灵活但是通常很慢，而且对于很多Class来说序列化结果比较大。
  • Kryo序列化: Spark可以使用Kryo库(第2版本)来进行更快速的序列化。Kryo比起Java序列化显著地更快而且结果空间利用更紧凑(通常是10倍以上)，但为了更好的性能，不支持所有Serializable的类型，而要求你提前注册将要在程序中使用的Class。

通过用SparkConf初始化任务并调用 conf.set(“spark.serializer”， “org.apache.spark.serializer.KryoSerializer”)， 你可以切换到Kryo序列化。这个配置设定的序列化框架不仅用于worker节点间的shuffling，也用于将RDD持续化到磁盘。不将Kryo作为默认序列化框架的唯一原因是它要求特定的注册步骤，但我们推荐在网络资源紧张的情况下使用它。

Spark为包含在 Twitter chill 的 AllScalaRegistar 库中的许多常用的Scala Class自动加载Kryo序列化框架。使用 registerKryoClasses 方法来注册你自定义的Class。

1
2
3
val conf = new SparkConf().setMaster(...).setAppName(...)
conf.registerKryoClasses(Array(classOf[MyClass1]， classOf[MyClass2]))
val sc = new SparkContext(conf)

Kryo官方文档描述了更多高级的注册选项，比如定制序列化代码。如果对象很大，你或许需要提高 spark.kryoserializr.buffer 配置。这个值需要足够大，可以放得下你需要序列化的最大的对象。最后，如果你不注册你定制的对象，Kryo仍然可以运作，但是它必须在每个对象存储全类名，这是很浪费的。

内存调优

内存调优主要有三个方面的考虑:对象使用的内存大小(你可能想要整个数据集都加载到内存)，访问这些对象的成本，还有垃圾回收的消耗(如果你需要大批量地创建和销毁对象)。
默认情况下，Java对象访问快，但每个属性很容易消耗原始数据2-5倍的空间。这是由几个原因造成的:

  • 每个不同的Java对象都有一个“对象头”，大约16字节，包含了Class的指针等信息。对于只存储了少量数据的对象（比如只有一个int属性），对象头可能会大于数据。
  • Java String对象有比起原始字符串需要多消耗40字节（因为数据以一个Char数组形式存放，要保存长度等额外的数据），并且每个字符是2个字节，因为String类型使用了UTF-16编码。所以一个10字符的字符串很容易就消耗了60字节。
  • 普遍的集合类，比如 HashMap 和 LinkedList，使用了链接形式的数据结构，这种数据结构每个成员都有一个“包装器”对象（比如 Map.Entry）。这个对象不仅有一个头部，而且有一个指向下个成员的指针（典型的是8字节）。
  • 原生类型的集合通常存储为“盒子”对象，比如 java.lang.Integer。

本章节会由Spark的内存管理概览开始，然后讨论一些特别的策略，这些策略用户可以在他或她的应用中更有效率地使用内存。特别地，我们会描述怎么样确定对象的内存使用和怎么去优化——通过改变你的数据结构，或者通过将数据以序列化方式存储。接下来我们再讨论Spark缓存优化和Java垃圾回收。

内存管理概览

Spark的内存使用分为两类：执行内存和存储内存。执行内存用于洗牌(shuffle)，连接(join)，排序(sort)和聚合(aggregation)，而存储内存指用于缓存和传输集群内部数据的内存。在Spark中，执行和存储共享同一内存区域（M区）。当不需要使用执行内存时，存储可以占据整个区域，反之亦然。需要的时候执行内存可能会驱逐存储内存，直到所有的存储内存使用降到某个阈值以下（R区）。换句话说，R区是M区的一个子分区，在这个分区内的缓存块不会被驱逐出内存。存储内存不会驱逐执行内存，否则实现就会过于复杂。

这样设计确保了几个重要的特性。第一，不需要缓存空间的应用可以使用整个内存作为执行内存，避免了无谓的内存溢出写到磁盘操作。第二，需要缓存空间的应用可以保留一个最小的存储空间（R区），这个空间内的数据块不会被驱逐出内存。最后，这个机制在不同工作负荷下提供了合理的“盒外”性能，不需要用户研究Spark内部怎么划分内存。

Spark提供了两个相关配置，但一般用户不应该需要调整它们，因为默认值在大多数情况都可以满足要求：

  • spark.memory.fraction 表示M区占据整个JVM堆（300MB）的比例，默认为0.6。留下40%的空间给用户数据结构，Spark内部元数据，以及防止过大记录导致OOM的预留空间。
  • spark.memory.storageFraction 表示R区占据M区空间的比例，默认为0.6。R区是M区中的存储区域，其中缓存的数据块不会被执行内存抢占。

spark.memory.fraction 的值应该被设置在JVM的老生代或者永生代的空间大小以内。详情参考进阶下文的GC调优讨论。

计算内存消耗

计算一个数据集需要的内存大小的最好的方式是，创建一个RDD并把它放进缓存，查看Spark Web界面的”Storage”页面。这个页面会告诉你这个RDD占用了多少内存。如果要估算某个对象的内存消耗，可以使用 SizeEstimator 的 estimate 方法。这无论在试验不同数据布局以减少内存使用的时候，还是在计算一个广播变量对于每个executor堆空间的占用大小的时候，都非常有用。

数据结构调优

减少内存消耗的第一个方式就是避免使用增加内存成本的Java特性，比如基于指针的数据结构和包装类。以下是几个具体的做法:

  1. 设计数据结构时优先使用对象数组和基本类型，而不是Java和Scala的标准集合(比如 HashMap)。Fastutil 库为基本类型提供了方便的集合类，并与Java标准库兼容。
  2. 尽量避免包含大量小对象和指针的复杂数据结构。
  3. 考虑使用数值类型的id或者枚举类型的key，避免使用字符串作为key。
  4. 如果内存不足32GB，设置JVM选项 -XX:+UseCompressedOops 将指针由默认8字节改为4字节。你可以将这些选项加到 spark-en.sh 中。

RDD序列化存储

当进行上述优化之后，对象仍旧太大而无法有效存储时，减少内存使用的一个更简单的办法是将它们序列化到磁盘存储，具体办法是使用RDD持续化API的序列化存储级别，比如 MEMORY_ONLY_SER。之后Spark会将每个RDD的分区存为一个大字节数组。这样做唯一的缺点是降低了访问效率，因为需要在访问时进行反序列化。如果你打算以序列化方式缓存数据，我们强烈推荐使用Kryo，因为它序列化的结果比较Java默认序列化小很多(当然也比原始Java对象小)。

垃圾回收(GC)调优

如果你在程序需要大量新建和销毁RDD操作的时候，JVM垃圾回收会成为一个问题(只是读取一个RDD然后操作多次不会产生这个问题)。Java需要将旧对象驱逐出内存来容纳新的对象，这时它会追踪所有的Java对象，找出其中不在使用的部分。这里的关键是垃圾回收的成本是和Java对象的数量成正比的，所以使用包含少量对象的数据结构(比如整形数组而不是 LinkedList)会显著减少这项成本。一个更好的办法是把对象序列化，就像上面描述的一样:这样每个RDD分区只有一个对象(一个字节数组)。在使用其他优化技巧之前，首先要尝试使用序列化存储来确定问题是不是出在垃圾回收上。
垃圾回收会造成问题的另外一种原因是任务的工作内存(执行任务需要的内存大小)和缓存在节点上的RDD竞争内存。我们接下来会讨论怎么通过控制给RDD分配的空间来减轻这种情况。

度量GC的影响

GC调优的第一步是收集数据，包括GC多久发生一次和花在GC上的时间。具体可以通过增加Java参数 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimeStamps。(详情见[配置教程]中的传递Java参数给Spark任务。)下次你的Spark任务执行的时候，你会看见每次GC发生时信息都会打印到worker的日志里面。注意这些日志会在集群的worker节点(在工作目录的stdout文件里)，而不是你的Driver节点。

进阶GC调优

为了进一步调优GC，我们首先需要理解一些关于JVM内存管理的基本信息:

  • Java堆空间分为新生代分区和老年代分区。新生代用于存放短可以生命周期的对象，而老年代则为长时间生存的对象准备。
  • 新生代进一步可以分为三个分区[Eden区，Survivor1，Survivor2]。
  • GC进程的简要描述:当Eden区满了，一个 minor GC会在Eden区启动，仍然生存的对象会从Eden取和Survivor1区拷贝到Survivor2区。两个Survivor区交换。如果一个对象足够老，或者Survior2满了，它会移到老年代分区。最后当老年代分区接近用满，full GC(全局GC)就会启动。

Spark GC调优的目标是确保只有长期生存的RDD存放在老年代区，而且新生代区足够存放短期生存的对象。这样就避免了启动full GC来收集任务执行时创建的临时对象。一些可能有帮助的步骤如下:

  • 收集GC统计数据，检查是否有过多的GC。如果在任务完成前full GC发生了多次，这意味着没有足够多的可用内存提供给该任务。
  • 如果有很多minor GC却没有很多major GC，分配更多内存给Eden区可以改善这个问题。你可以将Eden区的大小调为高于任务所需内存。如果Eden区的小大为E，你可以通过参数 -Xmn=4/3*E 设置新生代的大小。(增大为4/3倍是因为Survivor分区也需要空间。)
  • 在打印出来的GC统计信息中，如果老生代接近用满，降低 spark.memory.fraction 以减少用于缓存RDD的空间。缓存少一点对象总比拖慢任务执行要好。或者考虑减小新生代分区的大小也是可以的。这意味着将 -Xmn 调低，如果你已经按照上文做了的话。如果没有的话，尝试改变JVM的NewRatio参数。许多JVM默认将此参数设为2，意味这老年代占据堆大小的2/3。这个值应该要超过 spark.memory.fraction。
  • 尝试设置参数 -XX:+UseG1GC 应用G1GC垃圾收集器。在某些情况下，当垃圾回收成为瓶颈时它可以提高性能。注意在堆空间比较大的情况下，使用 –XX:G1HeapRegionSize 参数提高G1分区大小是很重要的。
  • 举例，如果你的任务从HDFS读取数据，可以通过读取的数据块大小估算任务使用的内存大小。注意解压后的数据块大小通常是原来大小的2至3倍。所以如果希望给3或4个任务分配工作空间，而且HDFS块大小为128M，我们可以估计Eden区大概需要 43128MB 的空间。
  • 更新设置后，监控GC的频率以及使用时间的变化。

根据经验我们认为GC调优的效果取决于具体应用和可用内存大小。网上有更多关于调优参数的描述，不过是高层的调优，以管理full GC发生的频率来帮助减轻GC成本。Executor的GC调优可以通过设置任务配置中的 spark.executor.extraJavaOptions 来指定。

其他考虑因素

并行度

除非你为每个操作设置足够高的并行度，否则集群资源不会被充分利用。Spark自动根据文件大小决定启动多少个map任务(虽然你可以通过可选参数或者 SparkContext.textFile等方法控制)。对于分布式的reduce操作，比如 groupByKey 和 reduceByKey，并行度是最大父RDD的分区数。你可以将并行度作为第二参数传递给Spark(见spark.PairRDDFunctions文档)，或者设置配置属性 spark.default.parallelism 来改变默认值。总的来说，我们推荐集群内每个CPU执行2至3个任务。

Reduce任务的内存使用

有时发生内存溢出错误并不是因为内存放不下RDD，而是其中一个task处理的数据集太大了，比如在 groupByKey 操作中就可能出现这种情况。Spark的shuffle操作(sortByKey， groupByKey， reduceByKey， join等等)为了进行分组操作，在每个task中都新建了一个哈希表。最简单的办法是提高并行度，然后每个task的输入都会变小。Spark执行时长200ms的短任务很有效率，因为它在许多task中复用了一个executor JVM，并且每个task的启动成本都较低，所以你可以安全地将并行度提高到集群cpu核数以上。

广播大变量

使用SparkContext中的广播函数可以显著减小每个序列化task的大小，还有在集群上启动任务的成本。如果任务需要从Driver程序获取大对象(比如静态的扫表)，你可以考虑将这个对象转变为广播变量。Spark将每个task序列化后的大小打印在master上，你可以根据这个来判断task是不是太大。通常来说task大于20KB就可能需要优化。

数据本地性

数据本地性对Spark任务的性能有重要影响。如果数据和处理它的代码在一起，计算会快一些。不过如果数据和代码是分开的，那么其中一个必须移动到另外一个那里。一般来说，移动序列化后的代码比移动一个数据块要快，因为代码远比数据小。Spark围绕数据本地性的一般原则来建立调度策略。
数据本地性指数据离对应的代码多近。有几个基于数据当前位置的本地性，从近到远如下:

  • PROCESS_LOCAL(进程本地) 数据就在代码所在的JVM里，这是最好的数据本地性。
  • NODE_LOCAL(节点本地) 数据在同一个节点上。例如在同一个节点的HDFS上，或者在同一节点的另外一个executor上。这比PROCESS_LOCAL稍慢，因为数据要跨进程传输。
  • NO_PREF(无偏好) 数据在所有节点的访问都一样快，没有本地性偏好。
  • RACK_LOCAL(机架本地) 数据在同一个机架的节点上。数据在相同机架的不同节点上所以需要网络传输。
  • ANY(任何) 数据在网络上其他地方，但不在同一机架上。

Spark倾向于以最好的数据本地性调度任务，但并不总能做到。在所有空闲的executor上都没有未处理数据的情况下，Spark将本地性要求放低。有两个方法：a)等待某个在工作的CPU空闲下来，可以在同一节点启动新的任务。b)马上在远端的节点启动新任务，但是需要将数据传输过去。

Spark典型的处理是等一会，希望有CPU资源释放。一旦等待超时，它开始将数据移动到远端的空闲CPU。不同级别间的等待时长可以分别设置或者同一设置。详情见配置页面的 spark.locality 参数。如果任务执行时间长或数据的本地性差，你应该调高这些时长，不过默认值同样也可以工作。

总结

这是一个简短的教程，指出了调优Spark时你应该知道的主要的关注点–最重要的是，数据序列化和内存调优。对于大多数程序来说，切换到Kryo序列化和将数据序列化存储会解决大部分常见的性能问题。欢迎在Spark邮件列表提出关于其他调优经验的问题。

本文是原创文章，转载请注明：时间与精神的小屋 - [译]Spark调优教程中文版